/**
 * 
 */
package com.netflexitysolutions.amazonws.sdb.orm.internal;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.netflexitysolutions.amazonws.sdb.Domain;
import com.netflexitysolutions.amazonws.sdb.SimpleDB;
import com.netflexitysolutions.amazonws.sdb.SimpleDBException;
import com.netflexitysolutions.amazonws.sdb.orm.PersistenceException;
import com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession;
import com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSessionFactory;
import com.netflexitysolutions.amazonws.sdb.orm.cache.Level2Cache;
import com.netflexitysolutions.amazonws.sdb.orm.filter.Predicate;
import com.netflexitysolutions.amazonws.sdb.orm.internal.operations.Count;
import com.netflexitysolutions.amazonws.sdb.orm.internal.operations.Delete;
import com.netflexitysolutions.amazonws.sdb.orm.internal.operations.Find;
import com.netflexitysolutions.amazonws.sdb.orm.internal.operations.FindAll;
import com.netflexitysolutions.amazonws.sdb.orm.internal.operations.LoadManyToMany;
import com.netflexitysolutions.amazonws.sdb.orm.internal.operations.LoadManyToOne;
import com.netflexitysolutions.amazonws.sdb.orm.internal.operations.LoadOneToMany;
import com.netflexitysolutions.amazonws.sdb.orm.internal.operations.Save;
import com.netflexitysolutions.amazonws.sdb.orm.internal.operations.Select;
import com.netflexitysolutions.amazonws.sdb.orm.metadata.IdMetadata;
import com.netflexitysolutions.amazonws.sdb.orm.metadata.ItemMetadata;
import com.netflexitysolutions.amazonws.sdb.orm.metadata.SimpleDBMapping;

/**
 * @author netflexity
 *
 */
public class SimpleDBSessionImpl implements SimpleDBSession{
	
	private static Log logger = LogFactory.getLog(SimpleDBSessionImpl.class.getName());
	
	private Map<String, Domain> domainsByName = Collections.synchronizedMap(new HashMap<String, Domain>());
	private SimpleDBSessionFactory factory;
	
	/**
	 * @param factory
	 */
	public SimpleDBSessionImpl(SimpleDBSessionFactory factory){
		assert(factory != null);
		this.factory = factory;
	}

    /* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#populate(java.lang.Object)
	 */
	public <T> void populate(T record) {
		Callable<Void> call = null;
		try {
			// Load many-to-one.
			call = new LoadManyToOne<T>(this, record);
			call.call();
        }
        catch (Exception e) {
            throw new PersistenceException(e);
        }
        
        try {
			// Load one-to-many.
			call = new LoadOneToMany<T>(this, record);
			call.call();
        }
        catch (Exception e) {
            throw new PersistenceException(e);
        }
        
        try {
			// Load many-to-many.
			call = new LoadManyToMany<T>(this, record);
			call.call();
        }
        catch (Exception e) {
            throw new PersistenceException(e);
        }
	}

	/* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#find(java.lang.Class, java.lang.String)
     */
    public <T> T find(Class<T> clazz, String id) {
        assert(id != null);
        try {
        	Find<T> find = new Find<T>(this, clazz, id);
        	return find.call();
        }
        catch (SimpleDBException e) {
            throw new PersistenceException(e);
        }
        catch (Exception e) {
	        throw new PersistenceException(e);
	    }
    }
    
    /* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#findAll(java.lang.Class, java.lang.String[])
     */
    public <T> Set<T> findAll(Class<T> clazz, String ... orderBy){
    	try {
        	FindAll<T> findAll = new FindAll<T>(this, clazz);
        	findAll.setOrderBy(orderBy);
        	return findAll.call();
        }
        catch (SimpleDBException e) {
            throw new PersistenceException(e);
        }
        catch (Exception e) {
	        throw new PersistenceException(e);
	    }
    }
    
    /* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#findAsync(java.lang.Class, java.lang.String)
     */
    public <T> Future<T> findAsync(Class<T> clazz, String id) {
        Future<T> future = factory.getQueryExecutor().submit(new Find<T>(this, clazz, id));
        return future;
    }
    
    /* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#findAllAsync(java.lang.Class, java.lang.String[])
	 */
	public <T> Future<Set<T>> findAllAsync(Class<T> clazz, String... orderBy) {
		FindAll<T> findAll = new FindAll<T>(this, clazz);
		findAll.setOrderBy(orderBy);
		Future<Set<T>> future = factory.getQueryExecutor().submit(findAll);
        return future;
	}

	/* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#select(com.netflexitysolutions.amazonws.sdb.orm.filter.Predicate, java.lang.String[])
	 */
	public <T> List<T> select(Predicate<T> predicate, String... orderBy) {
		try {
        	Select<T> select = new Select<T>(this, predicate, orderBy);
        	return select.call();
        }
        catch (SimpleDBException e) {
            throw new PersistenceException(e);
        }
        catch (Exception e) {
	        throw new PersistenceException(e);
	    }
	}
	
	/* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#selectAsync(com.netflexitysolutions.amazonws.sdb.orm.filter.Predicate, java.lang.String[])
	 */
	public <T> Future<List<T>> selectAsync(Predicate<T> predicate, String... orderBy) {
		Future<List<T>> future = factory.getQueryExecutor().submit(new Select<T>(this, predicate, orderBy));
		return future;
	}

	/* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#save(java.lang.Object)
     */
    public <T> void save(T object){
	    try {
	        new Save<T>(this, object).call();
	    } 
	    catch (SimpleDBException e) {
	        throw new PersistenceException(e);
	    } 
	    catch (Exception e) {
	        throw new PersistenceException(e);
	    }
    }
    
    /* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#saveAsync(java.lang.Object)
     */
    public <T> Future<T> saveAsync(T object) {
        Future<T> future = factory.getManipulationExecutor().submit(new Save<T>(this, object));
        return future;
    }
    
    /* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#remove(java.lang.Object)
     */
    public <T> void remove(T object) {
        if (object == null) return;
        try {
            Delete<T> delete = new Delete<T>(this, object);
            delete.call();
        } 
        catch (Exception e) {
            throw new PersistenceException(e);
        }
    }
    
    /* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#remove(java.lang.Class, java.lang.String)
     */
    public <T> void remove(Class<T> clazz, String id) {
        if (StringUtils.isBlank(id)) return;
        try {
            Delete<T> delete = new Delete<T>(this, clazz, id);
            delete.call();
        } 
        catch (Exception e) {
            throw new PersistenceException(e);
        }
    }
    
    /* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#removeAsync(java.lang.Object)
     */
    public <T> Future<Void> removeAsync(T record) {
        Future<Void> future = factory.getManipulationExecutor().submit(new Delete<T>(this, record));
        return future;
    }
    
    /* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#removeAsync(java.lang.Class, java.lang.String)
     */
    public <T> Future<Void> removeAsync(Class<T> clazz, String id) {
        Future<Void> future = factory.getManipulationExecutor().submit(new Delete<T>(this, clazz, id));
        return future;
    }
    
    /* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#count(com.netflexitysolutions.amazonws.sdb.orm.filter.Predicate)
	 */
	public <T> Integer count(Predicate<T> predicate) {
		try {
        	Count<T> count = new Count<T>(this, predicate);
        	return count.call();
        }
        catch (SimpleDBException e) {
            throw new PersistenceException(e);
        }
        catch (Exception e) {
	        throw new PersistenceException(e);
	    }
	}

	/* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#countAsync(com.netflexitysolutions.amazonws.sdb.orm.filter.Predicate)
	 */
	public <T> Future<Integer> countAsync(Predicate<T> predicate) {
		// TODO Auto-generated method stub
		return null;
	}

	/* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#avg(com.netflexitysolutions.amazonws.sdb.orm.filter.Predicate)
	 */
	public <T> int avg(Predicate<T> predicate) {
		// TODO Auto-generated method stub
		return 0;
	}

	/* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#avgAsync(com.netflexitysolutions.amazonws.sdb.orm.filter.Predicate)
	 */
	public <T> Future<Integer> avgAsync(Predicate<T> predicate) {
		// TODO Auto-generated method stub
		return null;
	}

	/* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#max(com.netflexitysolutions.amazonws.sdb.orm.filter.Predicate)
	 */
	public <T> int max(Predicate<T> predicate) {
		// TODO Auto-generated method stub
		return 0;
	}

	/* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#maxAsync(com.netflexitysolutions.amazonws.sdb.orm.filter.Predicate)
	 */
	public <T> Future<Integer> maxAsync(Predicate<T> predicate) {
		// TODO Auto-generated method stub
		return null;
	}

	/* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#min(com.netflexitysolutions.amazonws.sdb.orm.filter.Predicate)
	 */
	public <T> int min(Predicate<T> predicate) {
		// TODO Auto-generated method stub
		return 0;
	}

	/* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#minAsync(com.netflexitysolutions.amazonws.sdb.orm.filter.Predicate)
	 */
	public <T> Future<Integer> minAsync(Predicate<T> predicate) {
		// TODO Auto-generated method stub
		return null;
	}

	/* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#find(java.lang.String, java.lang.String, java.lang.Object)
	 */
	public <T> List<T> find(String query, String paramName, Object paramValue) {
		// TODO Auto-generated method stub
		return null;
	}

	/* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#find(java.lang.String, java.lang.String[], java.lang.Object[])
	 */
	public <T> List<T> find(String query, String[] paramNames,
			Object[] paramValues) {
		// TODO Auto-generated method stub
		return null;
	}

	/* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#getId(java.lang.Object)
     */
    public <T> Object getId(T record) {
    	assert(record != null);
    	ItemMetadata itemMetadata = getMetadata(record.getClass());
    	IdMetadata idMetadata = itemMetadata.getId();
    	return idMetadata.getFieldValue(record);
    }
    
    /* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#getMetadata(java.lang.Class)
     */
    public <T>ItemMetadata getMetadata(Class<T> clazz){
    	return factory.getMapping().getItemMetadataByClass(clazz);
    }
    
    /* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#getDomainName(java.lang.Class)
     */
    public <T>String getDomainName(Class<T> clazz){
    	ItemMetadata itemMetadata = getMetadata(clazz);
    	assert(itemMetadata != null);
    	return itemMetadata.getDomainName();
    }
    
    /* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#getDomain(java.lang.Class)
     */
    public <T> Domain getDomain(Class<T> clazz){
    	return getDomain(getDomainName(clazz));
    }
    
    /* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#getDomain(java.lang.String)
     */
    public Domain getDomain(String domainName) {
    	assert(domainName != null);
        synchronized(domainsByName) {
        	if(domainsByName.isEmpty()){
        		loadDomains();
        	}
        	
        	// If still no domains, lets try to create one.
        	if(domainsByName.isEmpty() || !domainsByName.containsKey(domainName)){
        		return createDomain(domainName);
        	}
        }
        return domainsByName.get(domainName);
    }
    
    /**
     * @param domainName
     * @return
     */
    public Domain createDomain(String domainName) {
    	if(logger.isDebugEnabled()){
			logger.debug("Creating domain: " + domainName);
		}
        try {
            Domain domain = getDataSource().createDomain(domainName);
            synchronized (domainsByName) {
            	domainsByName.put(domain.getName(), domain);
			}
            return domain;
        } 
        catch (SimpleDBException e) {
            throw new PersistenceException("Could not create SimpleDB domain:" + domainName, e);
        }
    }

    /**
     * @param domainName
     * @return
     */
    public Domain getOrCreateDomain(String domainName) {
        Domain domain = getDomain(domainName);
        if (domain == null) {
        	domain = createDomain(domainName);
        }
        return domain;
    }
    
    /**
     * @param <T>
     * @param c
     * @return
     */
    public <T>Domain getOrCreateDomain(Class<T> c) {
        return getOrCreateDomain(getDomainName(c));
    }

    /**
     * Load or reload domains from Amazon SimpleDB account.
     * If method gets called, domains will be reloaded no matter if already exist.
     */
    private void loadDomains() {
    	synchronized(domainsByName) {
	        try {
	            if(logger.isDebugEnabled()){
	            	logger.debug("Loading all domains...");
	            }
	            Iterator<List<Domain>> domainIterator = getDataSource().getDomains();
	            while (domainIterator.hasNext()) {
					List<Domain> domainsList = domainIterator.next();
					addDomains(domainsList);
				}
	        }
	        catch (SimpleDBException e) {
	            throw new PersistenceException(e);
	        }
    	}
    }

    /**
     * @param domainsList
     */
    private void addDomains(List<Domain> domainsList) {
    	synchronized(domainsByName) {
	        for (Domain domain : domainsList) {
	        	domainsByName.put(domain.getName(), domain);
	        }
    	}
    }
    
	/* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#cacheGet(java.lang.Class, java.lang.String)
     */
    public <T> T cacheGet(Class<T> clazz, String id) {
    	if(logger.isDebugEnabled()){
        	logger.debug("Trying to get object of type " + clazz.getSimpleName() + " from 2nd level cache, identified as " + id);
        }
        String cacheName = cacheName(clazz);
        Level2Cache<T> cache = factory.getCache(cacheName);
        T record = null;
        if (cache != null) {
            record = cache.get(id);
            if (record != null) {
                if(logger.isDebugEnabled()){
                	logger.debug("Retrieved object from cache of type " + clazz.getSimpleName() + ", identified as " + id);
                }
            }
        }
        return record;
    }
    
    /* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#cacheGetAll(java.lang.Class)
     */
    public <T> List<T> cacheGetAll(Class<T> clazz) {
    	if(logger.isDebugEnabled()){
        	logger.debug("Trying to get all objects of type " + clazz.getSimpleName() + " from 2nd level cache");
        }
        String cacheName = cacheName(clazz);
        Level2Cache<T> cache = factory.getCache(cacheName);
        if (cache != null) {
        	List<String> keys = cache.getKeys();
        	if(!keys.isEmpty()){
        		List<T> cachedRecords = new ArrayList<T>(keys.size());
        		for (String key : keys) {
        			T cachedRecord = cache.get(key);
        			if(cachedRecord != null){
        				cachedRecords.add(cachedRecord);
        			}
				}
        		if(logger.isDebugEnabled()){
                	logger.debug("Retrieved (" + cachedRecords.size() + ") objects from cache of type " + clazz.getSimpleName());
                }
        		return cachedRecords;
        	}
        }
        return null;
    }
    
    /* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#cachePut(java.lang.Object)
     */
    public <T> void cachePut(T record) {
    	Object id = getId(record);
    	assert(id != null && StringUtils.isNotBlank(String.valueOf(id)));
    	
    	String cacheName = cacheName(record.getClass());
    	if(logger.isDebugEnabled()){
        	logger.debug("Caching object of type " + record.getClass().getSimpleName() + ", identified as " + id);
        }
    	
    	Level2Cache<T> cache = factory.getCache(cacheName);
        if (cache != null) {
            cache.set(String.valueOf(id), record);
        }
    }

    /* (non-Javadoc)
     * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#cacheRemove(java.lang.Class, java.lang.String)
     */
    public <T> void cacheRemove(Class<T> clazz, String id) {
        String cacheName = cacheName(clazz);
        if(logger.isDebugEnabled()){
        	logger.debug("Removing object from cache of type " + clazz.getSimpleName() + ", identified as " + id);
        }
    	
    	Level2Cache<T> cache = factory.getCache(cacheName);
        if (cache != null) {
            cache.remove(id);
        }
    }

    /**
     * @param <T>
     * @param clazz
     * @return
     */
    protected <T> String cacheName(Class<T> clazz) {
        ItemMetadata itemMetadata = getMetadata(clazz);
        assert(itemMetadata != null);
        return itemMetadata.getClassName();
    }

	/**
	 * @return the mapping
	 */
	public SimpleDBMapping getMapping() {
		return factory.getMapping();
	}

	/**
	 * @return the dataSource
	 */
	public SimpleDB getDataSource() {
		return factory.getDataSource();
	}

	/* (non-Javadoc)
	 * @see com.netflexitysolutions.amazonws.sdb.orm.SimpleDBSession#getFactory()
	 */
	public SimpleDBSessionFactory getFactory() {
		return factory;
	}
}
